package self;

import redis.clients.jedis.*;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;

/**
 * Created by 刘万祥 on 2017/5/13 0013.
 */
public class Chapter04 {

    private final String inventory_key_prefix_s = "inventory:";
    private final String market_key_z = "market:";
    private final String user_key_prefix_h = "users:";
    private final String login_key_h = "users:";
    private final String recent_key_z = "recent:";
    private final String viewed_key_prefix_z = "viewed:";

    public static void main(String[] args) {

        new Chapter04().run();
    }

    private void run() {

        Jedis conn = new Jedis("localhost");
        conn.select(15);

        testListItem(conn);
        testPurchaseItem(conn);
//        非事务型流水线
        testPipelineUpdateToken(conn);
    }

    private void testPipelineUpdateToken(Jedis conn) {
        Pipeline pipeline = conn.pipelined();
        pipeline.multi();

        String token = UUID.randomUUID().toString();
        String user = "userX";
        pipeline.hset(login_key_h, token, user);

        long timestamp = System.currentTimeMillis() / 1000;
        pipeline.zadd(recent_key_z, timestamp, token);

        String item = "itemX";
        pipeline.zadd(viewed_key_prefix_z + token, timestamp, item);
        Long viewedSize = pipeline.zcard(viewed_key_prefix_z + token).get();
        long end = viewedSize > 25 ? 25 : viewedSize;
        pipeline.zremrangeByRank(viewed_key_prefix_z + token, 0, end);
        pipeline.zincrby(viewed_key_prefix_z + token, -1, item);

        pipeline.exec();
    }

    private void testPurchaseItem(Jedis conn) {
        String userY = "userY";
        conn.hset(user_key_prefix_h + userY, "funds", "110");
        Map<String, String> userYs = conn.hgetAll(user_key_prefix_h + userY);
        for (Map.Entry<String,String> entry : userYs.entrySet()){
            System.out.println("  " + entry.getKey() + ": " + entry.getValue());
        }

        String userX = "userX";
        String itemId = "itemX";
        purchaseItem(conn, userY, itemId, userX);
    }

    private boolean purchaseItem(Jedis conn, String buyerId, String itemId, String sellerId) {
        String buyer = user_key_prefix_h + buyerId;
        String seller = user_key_prefix_h + sellerId;
        String item = itemId + "." + sellerId;
//        等下更新买家的商品库存
        String buyerInventory = inventory_key_prefix_s + buyerId;

        long end = System.currentTimeMillis() + 10_0000;

        while (System.currentTimeMillis() < end) {
            conn.watch(market_key_z, buyer);

            double marketPrice = conn.zscore(market_key_z, item);
            double funds = Double.parseDouble(conn.hget(buyer, "funds"));
            if (marketPrice > funds) {
                conn.unwatch();
                return false;
            }

            Transaction trans = conn.multi();
//            价格变化
            trans.hincrBy(seller, "funds", (long) marketPrice);
            trans.hincrBy(buyer, "funds", (long) -marketPrice);
//            库存变化
            trans.zrem(market_key_z, item);
            trans.sadd(buyerInventory, item);

            List<Object> objects = trans.exec();
            if (objects == null) {
                continue;
            }

            return true;
        }

        return false;
    }

    private void testListItem(Jedis conn) {
        String seller = "userX";
        String item = "itemX";

        Long addResult = conn.sadd(inventory_key_prefix_s + seller, item);
        System.out.println(addResult);

        boolean listSuccess = listItem(conn, seller, item, 10);
        System.out.println(listSuccess);

        Set<Tuple> markets = conn.zrangeWithScores(market_key_z, 0, -1);
        for (Tuple market : markets) {
            System.out.println(market.getElement() + "   ,   " + market.getScore());
        }
    }

    private boolean listItem(Jedis conn, String sellerId, String itemId, double price) {
        String inventory = inventory_key_prefix_s + sellerId;
        String item = itemId + "." + sellerId;

        long end = System.currentTimeMillis() + 5000;
        while (System.currentTimeMillis() < end) {

            conn.watch(inventory);
//            判断卖家的商品是否还存在
            if (!conn.sismember(inventory, itemId)) {
                conn.unwatch();
                return false;
            }

            Transaction trans = conn.multi();
            trans.zadd(market_key_z, price, item);
            trans.srem(inventory,itemId);

            List<Object> objects = trans.exec();
//            在执行时候,watch 到 key 数据发生变化
            if (objects == null) {
                continue;
            }

            return true;
        }

        return false;
    }
}













